﻿// This code is released under the BSD license.
namespace PugLib.IO
{
    // Stephen Toub is the original author
    using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Threading;

    /// <summary>Parallelized pipeline for processing streams.</summary>
    public sealed class StreamPipeline : IDisposable
    {
        /// <summary>The list of stream processing filters provided by the user.</summary>
        private readonly Action<Stream, Stream>[] _filters;

        /// <summary>The blocking streams used between the filters.</summary>
        private List<BlockingStream> _blockingStreams;

        /// <summary>Initializes a StreamPipeline.</summary>
        /// <param name="filters">The filters used to process the input stream into the output stream.</param>
        public StreamPipeline(params Action<Stream, Stream>[] filters)
        {
            // Validate all arguments.
            if (filters == null)
            {
                throw new ArgumentNullException("filters");
            }
            if (filters.Length == 0 || Array.IndexOf(filters, null) >= 0)
            {
                throw new ArgumentException("filters empty");
            }

            // Store the arguments
            _filters = filters;

            // Create any necessary blocking streams.  We need one between each filter.
            _blockingStreams = new List<BlockingStream>(_filters.Length - 1);
            for (int i = 0; i < filters.Length - 1; i++)
            {
                _blockingStreams.Add(new BlockingStream());
            }
        }

        #region IDisposable Members

        /// <summary>Releases all resources used by the pipeline.</summary>
        public void Dispose()
        {
            // Dispose all of the blocking queues.
            if (_blockingStreams != null)
            {
                foreach (BlockingStream stream in _blockingStreams)
                {
                    stream.Dispose();
                }
                _blockingStreams = null;
            }
            GC.SuppressFinalize(this);
        }

        #endregion

        /// <summary>Runs the pipeline.</summary>
        /// <param name="input">The input stream that feeds into the pipeline.</param>
        /// <param name="output">The output stream that feeds out of the pipeline.</param>
        /// <remarks>This method is synchronous: it will not return until the pipeline is complete.</remarks>
        public void Run(Stream input, Stream output)
        {
            // Validate all arguments.
            if (_blockingStreams == null)
            {
                throw new ObjectDisposedException(GetType().Name);
            }
            if (input == null)
            {
                throw new ArgumentNullException("input");
            }
            if (!input.CanRead)
            {
                throw new ArgumentException("input is unreadable");
            }
            if (output == null)
            {
                throw new ArgumentNullException("output");
            }
            if (!output.CanWrite)
            {
                throw new ArgumentException("output is unwritable");
            }

            // Create a delegate to execute each stage of the pipeline.  The initial
            // stage takes the user-provided input stream, runs it through the first filter,
            // and outputs it to the first blocking queue.  The second stage takes the first
            // blocking queue, runs it through the second filter, and outputs it to the second
            // blocking queue.  Etc.  The final stage uses the last filter and pipes the output
            // to the user-provided output stream.
            ThreadStart lastStage = null;
            for (int i = 0; i < _filters.Length; i++)
            {
                Stream stageInput = i == 0 ? input : _blockingStreams[i - 1];
                Stream stageOutput = i == _filters.Length - 1 ? output : _blockingStreams[i];
                Action<Stream, Stream> filter = _filters[i];
                ThreadStart stage = delegate
                                        {
                                            filter(stageInput, stageOutput);
                                            if (stageOutput is BlockingStream)
                                            {
                                                ((BlockingStream)stageOutput).SetEndOfStream();
                                            }
                                        };
                if (i < _filters.Length - 1)
                {
                    Thread t = new Thread(stage);
                    t.IsBackground = true;
                    t.Start();
                }
                else
                {
                    lastStage = stage;
                }
            }
            if (lastStage != null)
            {
                lastStage(); // Won't return until end of stream.
            }
        }
    }
}